package com.intretech.tj.device;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.intretech.tj.global.parent.TriggerParent;
import com.intretech.tj.mqtt.MQTTVerticle;
import com.intretech.tj.mqtt.trigger.CloseTrigger;
import com.intretech.tj.mqtt.trigger.TopicMessageTrigger;
import io.smallrye.mutiny.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import static com.intretech.tj.device.OutLog.JSONError;
import static com.intretech.tj.global.tools.JSONTool.toJSONString;

@Slf4j
@ApplicationScoped
public class DeviceVerticle extends AbstractVerticle {
    private final OutLog outLog;
    private final DeviceConfig config;
    private final MQTTVerticle mqttVerticle;
    private final TopicMessageTrigger topicMessageTrigger;

    private final CloseTrigger closeTrigger;

    private final Map<String, Device> deviceMap = new ConcurrentHashMap<>();

    @Inject
    public DeviceVerticle(OutLog outLog, DeviceConfig config, MQTTVerticle mqttVerticle, TopicMessageTrigger topicMessageTrigger, CloseTrigger closeTrigger) {
        this.outLog = outLog;
        this.config = config;
        this.mqttVerticle = mqttVerticle;
        this.topicMessageTrigger = topicMessageTrigger;
        this.closeTrigger = closeTrigger;
    }

    @Override
    public void start(Promise<Void> startPromise) {
        topicMessageTrigger.setTrigger(config.Device(), outLog.getServerName(), (idAndMessage) -> {
            ObjectMapper objectMapper = new ObjectMapper();
            try {
                Device device = objectMapper.readValue(idAndMessage.message().payload().toString(), Device.class);
                device.setMqtt(idAndMessage.clientID());
                update(device);
            } catch (JsonProcessingException e) {
                log.error(outLog.getMessage(JSONError));
            }
        });

        closeTrigger.setTrigger(TriggerParent.ALL,outLog.getServerName(),(clientID)->{
            List<String> keys = new ArrayList<>();
            deviceMap.values().forEach(device -> {
                if (Objects.equals(device.getMqtt(), clientID)) {
                    keys.add(device.getKey());
                }
            });
            keys.forEach(deviceMap::remove);
        });
    }

    /**
     * 更新<br/>
     * 留出开放函数
     *
     * @param device 设备
     */
    public void update(Device device) {
        deviceMap.put(device.getKey(), device);
//        log.info(outLog.getMessage(Update, device.getKey(), device.getState(), device.getError()));
        Map<String, List<Device>> devicesByType = deviceMap.values().stream()
                .collect(Collectors.groupingBy(item->item.getType().getTitle()));
        mqttVerticle.retain(config.All(), 0, toJSONString(toJSONString(devicesByType)));
    }
}
